/*
 * Copyright (c) Jipzingking 2016.
 */

package sdfs.datanode;

import sdfs.protocol.IDataNodeProtocol;
import sdfs.util.LRULinkedHashMap;

import java.io.*;
import java.util.UUID;

public class DataNode implements IDataNodeProtocol {
    /**
     * The block size may be changed during test.
     * So please use this constant.
     */
    public static final int BLOCK_SIZE = 128 * 1024;
    public static final int DATA_NODE_PORT = 4341;
    //    put off due to its difficulties
//        private final Map<UUID, Set<Integer>> uuidReadonlyPermissionCache = new HashMap<>();
//        private final Map<UUID, Set<Integer>> uuidReadwritePermissionCache = new HashMap<>();

    private LRULinkedHashMap<Integer,byte[]> lru;
    public DataNode(int fileDataBlockCacheSize){
        lru = new LRULinkedHashMap<>(fileDataBlockCacheSize);
    }

    /**
     * Read data from a block.
     * It should be redirect to [blockNumber].block file
     *
     * @param fileUuid    the file uuid to check whether have permission to read or not. Put off to future lab.
     * @param blockNumber the block number to be read
     * @param offset      the offset on the block file
     * @param size        the total size to be read
     * @return the total number of bytes read into the buffer
     * @throws IndexOutOfBoundsException if offset less than zero, or offset+size larger than block size.
     * @throws FileNotFoundException     if the block is free (block file not exist)
     * @throws IllegalStateException     if uuid is illegal or has no permission on this file
     */
    @Override
    public byte[] read(UUID fileUuid, int blockNumber, int offset, int size) throws IndexOutOfBoundsException, IOException {
//        FileNode fileNode = NameNode.readonlyFile.get(fileUuid);
//        if (fileNode==null)throw new IllegalStateException();

        if (size < 0 ||(offset + size) > BLOCK_SIZE || offset < 0) throw new IndexOutOfBoundsException();

        String path = "src/data/" + blockNumber + ".block";
        File file = new File(path);
        if (!file.exists()) throw new FileNotFoundException("file " + path + " not found！！");
        FileInputStream fileInputStream = new FileInputStream(file);

        if (offset==0)
            if (lru.containsKey(blockNumber))
                return lru.get(blockNumber);

        long skip = fileInputStream.skip(offset);
        if (skip<= 0)
            throw new IOException();
        byte[] temp = new byte[size];
        int byteSize = fileInputStream.read(temp);//read byte to byte[]
        byte [] bytes = new byte[byteSize];
        System.arraycopy(temp,0,bytes,0,byteSize);
        fileInputStream.close();
        return bytes;
    }

    /**
     * Write data to a block.
     * It should be redirect to [blockNumber].block file
     *
     * @param fileUuid    the file uuid to check whether have permission to write or not. Put off to future lab.
     * @param blockNumber the block number to be written
     * @param offset      the offset on the block file
     * @param b           the buffer that stores the data
     * @throws IndexOutOfBoundsException if offset less than zero, or offset+size larger than block size.
     * @throws IllegalStateException     if uuid is illegal or has no permission on this file
     */
    @Override
    public void write(UUID fileUuid, int blockNumber, int offset, byte[] b) throws IndexOutOfBoundsException, IOException {
//        FileNode fileNode = NameNode.readwritePFile.get(fileUuid);
//        if (fileNode==null)throw new IllegalStateException("no permission to write "+fileUuid);

        if (offset < 0 ||(offset + b.length) > BLOCK_SIZE) throw new IndexOutOfBoundsException();

        String path = "src/data/" + blockNumber + ".block";
        File file = new File(path);
//        FileOutputStream fileOutputStream = new FileOutputStream(file,true);
//        fileOutputStream.write(b,0,b.length);
        FileOutputStream fileOutputStream = new FileOutputStream(file);
        fileOutputStream.write(b);
        fileOutputStream.flush();
        fileOutputStream.close();
        lru.put(blockNumber,b);
    }
}
